package com.yukefms.engine.running;

import com.yukefms.message.KafkaConfig;
import com.yukefms.message.MessageProducer;
import com.yukefms.rule.state.InActivatedState;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * @Description
 * @Author Ping
 * @Date 2021/7/14 8:44
 **/
public class ToKafkaListenerDecorator implements RunningEventListener {
    public static Logger logger = LogManager.getLogger(ToKafkaListenerDecorator.class);
    private RunningEventListener  listener ;
    private MessageProducer producer ;

    public ToKafkaListenerDecorator() {
    }

    public ToKafkaListenerDecorator(final RunningEventListener listener) {
        this.listener = listener;
    }

    public ToKafkaListenerDecorator(final RunningEventListener listener, final MessageProducer producer) {
        this.listener = listener;
        this.producer = producer;
    }

    @Override
    public void handle(RunningEvent event, RunningContext context) {
        listener.handle(event, context);
        if (event.getEventCode() != EventCode.ALL_FINISHED) {
            event.setCurrentScore(context.currentScore());
        }
        event.setTrialNum(context.getTrialoutCounter().getCurNum());
        logger.error("produced: {}", event.toString() ) ;
        this.producer.send(KafkaConfig.TOPIC_RESPONSE, event.toString());


    }

    public void setListener(final RunningEventListener listener) {
        this.listener = listener;
    }

    public void setProducer(final MessageProducer producer) {
        this.producer = producer;
    }
}
